package net.towee.server.rpc;

import java.io.IOException;
import java.io.InputStream;
import java.lang.reflect.Method;
import java.net.MalformedURLException;
import java.net.URL;
import java.nio.charset.Charset;
import java.nio.charset.CharsetDecoder;
import java.text.ParseException;
import java.util.HashMap;
import java.util.Map;

import javax.servlet.GenericServlet;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpSession;

import net.towee.server.authentication.SessionAuthManager;
import net.towee.server.rpc.services.RPCUtils;
import net.towee.server.sio.common.DisconnectReason;
import net.towee.server.sio.common.SocketIOException;
import net.towee.server.sio.server.SocketIOInbound;
import net.towee.server.sio.server.SocketIOServlet;
import net.towee.server.sio.server.SocketIOInbound.SocketIOOutbound;

import org.apache.mina.core.buffer.IoBuffer;
import org.apache.vysper.mina.codec.StanzaBuilderFactory;
import org.apache.vysper.xml.decoder.DocumentContentHandler;
import org.apache.vysper.xml.decoder.XMLElementListener;
import org.apache.vysper.xml.fragment.Renderer;
import org.apache.vysper.xml.fragment.XMLElement;
import org.apache.vysper.xml.sax.NonBlockingXMLReader;
import org.apache.vysper.xml.sax.impl.DefaultNonBlockingXMLReader;
import org.apache.vysper.xmpp.protocol.NamespaceURIs;
import org.apache.vysper.xmpp.protocol.SessionStateHolder;
import org.apache.vysper.xmpp.protocol.StreamErrorCondition;
import org.apache.vysper.xmpp.server.AbstractSessionContext;
import org.apache.vysper.xmpp.server.ServerRuntimeContext;
import org.apache.vysper.xmpp.server.SessionState;
import org.apache.vysper.xmpp.server.response.ServerErrorResponses;
import org.apache.vysper.xmpp.stanza.Stanza;
import org.apache.vysper.xmpp.writer.StanzaWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.xml.sax.SAXException;

import com.google.gwt.thirdparty.guava.common.base.Preconditions;
import com.google.gwt.user.client.rpc.IncompatibleRemoteServiceException;
import com.google.gwt.user.client.rpc.RpcTokenException;
import com.google.gwt.user.client.rpc.SerializationException;
import com.google.gwt.user.server.rpc.RPC;
import com.google.gwt.user.server.rpc.RPCRequest;
import com.google.gwt.user.server.rpc.SerializationPolicy;
import com.google.gwt.user.server.rpc.SerializationPolicyLoader;
import com.google.gwt.user.server.rpc.SerializationPolicyProvider;
import com.google.gwt.user.server.rpc.UnexpectedException;
import com.mycila.event.Dispatcher;
import com.mycila.event.Event;
import com.mycila.event.Subscriber;
import com.mycila.event.Topic;
import com.mycila.event.Topics;

/**
 * Basic servlet logic for xmpp and push methods
 * 
 * @author c58
 */
public abstract class ToweeSocketServlet extends SocketIOServlet implements SerializationPolicyProvider {
	private static final long serialVersionUID = 197413099255163742L;

	/**
	 * This socket connection provides parsing and processing request/response
	 * and XMPP stanzas.
	 * 
	 * @author c58
	 */
	private class SocketConnection extends AbstractSessionContext implements SocketIOInbound,
			StanzaWriter, Subscriber<Object[]>, XMLElementListener {

		// For XMPP stanza decoding
		private final Charset CHARSET = Charset.forName("UTF-8");
		private final CharsetDecoder CHARSET_DECODER = CHARSET.newDecoder();
		private final NonBlockingXMLReader xmlReader = new DefaultNonBlockingXMLReader();

		// Output socket
		private SocketIOOutbound outbound = null;

		// Intializaion flag
		private boolean init = false;

		// Permutation for GWT-RPC
		private String moduleBaseURL;
		private String strongName;
		
		private final HttpSession session;

		public SocketConnection(ServerRuntimeContext serverRuntimeContext, HttpSession session) {
			super(serverRuntimeContext, new SessionStateHolder());
			
			Preconditions.checkNotNull(serverRuntimeContext, "server runtime is null");
			Preconditions.checkNotNull(session, "session is null");
			
			DocumentContentHandler contentHandler = new DocumentContentHandler(new StanzaBuilderFactory());
			contentHandler.setListener(this);
			
			this.session = session;

			try {
				// we need to check the jabber:client/jabber:server NS
				// declarations
				xmlReader.setFeature(DefaultNonBlockingXMLReader.FEATURE_NAMESPACE_PREFIXES, true);
				// allow parser to restart XML stream
				xmlReader.setFeature(DefaultNonBlockingXMLReader.FEATURE_RESTART_ALLOWED, true);
				xmlReader
						.setProperty(DefaultNonBlockingXMLReader.PROPERTY_RESTART_QNAME, "stream:stream");
			} catch (SAXException e) {
				// should never happen
				throw new RuntimeException(e);
			}

			xmlReader.setContentHandler(contentHandler);
		}
		
		public HttpSession getHttpSession() {
			return session;
		}

		@Override
		public String getProtocol() {
			return null;
		}

		@Override
		public void onConnect(SocketIOOutbound outbound) {			
			LOG.info("SocketIO client connected");
			this.outbound = outbound;
		}

		@Override
		public void onDisconnect(DisconnectReason reason, String errorMessage) {
			LOG.info("SocketIO client disconnected "+reason);
			this.outbound = null;

			// End XMPP session
			endSession(SessionTerminationCause.CONNECTION_ABORT);

			// Remove subscriber from dispatcher
			dispatcher.unsubscribe( this );
		}

		@Override
		public void onMessage(String message) {
			LOG.info("Received message: "+message+" "+sessionStateHolder.getState());
			
			if (!init)
				initializeSocket(message);
			else
				processMessage(message);
		}

		private void initializeSocket(String message) {
			// Split message by |
			String[] parts = message.split("\\|", 2);

			// Check splitting result
			if (parts.length != 2)
				outbound.disconnect();

			// Set connection as initialized
			strongName = parts[0];
			moduleBaseURL = parts[1];
			init = true;

			// Set socket in thread storage
			threadSocket.set(this);
			initSubscribers();
		}

		private void processMessage(String message) {
			Preconditions.checkArgument(init);

			// Set socket in thread storage
			threadSocket.set(this);

			// Get message type
			String requestType = message.substring(0, 1);

			// If it's request
			if (requestType.equals("0"))
				try {
					LOG.info("Message was reauest/response. Process it.");
					processRequest(message.substring(1), outbound);
				} catch (SocketIOException e1) {
				} catch (SerializationException e1) {
				}

			else if (requestType.equals("2"))
				try {
					LOG.info("Message was XMPP stanza: "+message.substring(1));
					xmlReader.parse(IoBuffer.wrap(message.substring(1).getBytes(CHARSET.name())),
							CHARSET_DECODER);
				} catch (IOException e) {
					// should never happen since we read from a string
					throw new RuntimeException(e);
				} catch (SAXException e) {
					Stanza errorStanza = ServerErrorResponses.getStreamError(
							StreamErrorCondition.XML_NOT_WELL_FORMED, getXMLLang(),
							"Stanza not well-formed", null);
					write(errorStanza);
					endSession(SessionTerminationCause.STREAM_ERROR);
				}
		}

		/**
		 * {@inheritDoc}
		 */
		public StanzaWriter getResponseWriter() {
			return this;
		}

		/**
		 * {@inheritDoc}
		 */
		public void switchToTLS(boolean delayed, boolean clientTls) {
			// n/a
		}

		/**
		 * {@inheritDoc}
		 */
		public void setIsReopeningXMLStream() {
			// n/a
		}

		/**
		 * {@inheritDoc}
		 */
		public void element(XMLElement element) {
			LOG.info("Process parsed stanza: "+element.toString()+" "+sessionStateHolder.getState());
			
			// Set session as 
			if( sessionStateHolder.getState() != SessionState.AUTHENTICATED ) {
				setInitiatingEntity( sessionManager.getSessionJID( session ) );
				sessionStateHolder.setState(SessionState.AUTHENTICATED);
			}
			
			// on parsed stanzas
			serverRuntimeContext.getStanzaProcessor().processStanza(serverRuntimeContext, this,
					(Stanza) element, sessionStateHolder);
		}

		/**
		 * {@inheritDoc}
		 */
		public void write(Stanza stanza) {
			// handle stream open
			Renderer renderer = new Renderer(stanza);

			if ("stream".equals(stanza.getName())
					&& NamespaceURIs.HTTP_ETHERX_JABBER_ORG_STREAMS.equals(stanza.getNamespaceURI())) {
				// stream:stream and stream:features comes at the same time,
				// split them
				send(renderer.getOpeningElement());
				send(renderer.getElementContent());
			} else {
				send(renderer.getComplete());
			}
		}

		private void send(String xml) {
			LOG.info("Write stanza: "+xml);
			
			try {
				outbound.sendMessage("2"+xml);
			} catch (IOException e) {
				// communication with client broken, close session
				endSession(SessionTerminationCause.CONNECTION_ABORT);
			}
		}

		/**
		 * {@inheritDoc}
		 */
		public void close() {
		}

		/**
		 * Invoked when some event published
		 */
		@Override
		public void onEvent(Event<Object[]> event) throws Exception {
			Object[] source = event.getSource();
			
			// Do nothing if this publisher is subscriber too
			if(source[2] == this)
				return;
			
			// Get method name and result object
			Method method = (Method) source[0];
			Object result = source[1];
			
			LOG.info("Send result to method: "+method.getName());

			// Get serialization policy
			SerializationPolicy serializationPolicy = getSerializationPolicy(moduleBaseURL, strongName);

			// Encode result
			String responsePayload = RPC.encodeResponseForSuccess(method, result, serializationPolicy, 1);

			// Create response message
			responsePayload = "1"
					+ RPCUtils.hexToLength(Integer.toHexString(method.getName().length()), 2)
					+ method.getName() + responsePayload;
			
			LOG.info("Response payload to send: "+responsePayload);

			// Send response
			outbound.sendMessage(responsePayload);
		}
	}

	/*
	 * XMPP and socket processing part
	 */
	private final static Logger LOG = LoggerFactory.getLogger(ToweeSocketServlet.class);

	public static final String SERVER_RUNTIME_CONTEXT_ATTRIBUTE = "org.apache.vysper.xmpp.server.ServerRuntimeContext";

	private final ServerRuntimeContext serverRuntimeContext;

	@Override
	protected SocketIOInbound doSocketIOConnect(HttpServletRequest request, String[] protocols) {
		return new SocketConnection(serverRuntimeContext, request.getSession());
	}

	/*
	 * RPC request processing part
	 */

	/**
	 * Used by HybridServiceServlet.
	 */
	static SerializationPolicy loadSerializationPolicy(GenericServlet servlet, String moduleBaseURL,
			String strongName) {
		String modulePath = null;
		if (moduleBaseURL != null) {
			try {
				modulePath = new URL(moduleBaseURL).getPath();
			} catch (MalformedURLException ex) {
				// log the information, we will default
				servlet.log("Malformed moduleBaseURL: " + moduleBaseURL, ex);
			}
		}

		SerializationPolicy serializationPolicy = null;

		/*
		 * Check that the module path must be in the same web app as the servlet
		 * itself. If you need to implement a scheme different than this,
		 * override this method.
		 */
		if (modulePath == null) {
			String message = "ERROR: The module path requested, "
					+ modulePath
					+ ", is not in the same web application as this servlet. "
					+ "  Your module may not be properly configured or your client and server code maybe out of date.";
			servlet.log(message);
		} else {
			String serializationPolicyFilePath = SerializationPolicyLoader
					.getSerializationPolicyFileName(modulePath + strongName);

			// Open the RPC resource file and read its contents.
			InputStream is = servlet.getServletContext().getResourceAsStream(serializationPolicyFilePath);
			try {
				if (is != null) {
					try {
						serializationPolicy = SerializationPolicyLoader.loadFromStream(is, null);
						LOG.info("Serialization policy loaded successfully.");
					} catch (ParseException e) {
						servlet.log("ERROR: Failed to parse the policy file '"
								+ serializationPolicyFilePath + "'", e);
					} catch (IOException e) {
						servlet.log("ERROR: Could not read the policy file '"
								+ serializationPolicyFilePath + "'", e);
					}
				} else {
					String message = "ERROR: The serialization policy file '"
							+ serializationPolicyFilePath
							+ "' was not found; did you forget to include it in this deployment?";
					servlet.log(message);
				}
			} finally {
				if (is != null) {
					try {
						is.close();
					} catch (IOException e) {
						// Ignore this error
					}
				}
			}
		}

		return serializationPolicy;
	}

	/**
	 * Local thread storage for GWT permutation
	 */
	private final transient ThreadLocal<SocketConnection> threadSocket;

	/**
	 * A cache of moduleBaseURL and serialization policy strong name to
	 * {@link SerializationPolicy}.
	 */
	private final Map<String, SerializationPolicy> serializationPolicyCache = new HashMap<String, SerializationPolicy>();

	/**
	 * The implementation of the service.
	 */
	private final Object delegate;

	/**
	 * Dispatcher of push events
	 */
	private final Dispatcher dispatcher;

	/**
	 * Session manager for getting account JID
	 */
	protected final SessionAuthManager sessionManager;

	/**
	 * The default constructor used by service implementations that extend this
	 * class. The servlet will delegate AJAX requests to the appropriate method
	 * in the subclass.
	 */
	public ToweeSocketServlet(Dispatcher dispatcher, SessionAuthManager sessionManager, ServerRuntimeContext xmppServer) {
		this.delegate = this;
		this.threadSocket = new ThreadLocal<SocketConnection>();
		this.dispatcher = dispatcher;
		this.sessionManager = sessionManager;
		this.serverRuntimeContext = xmppServer;
	}
	
	protected HttpSession getSession() {
		return threadSocket.get().getHttpSession();
	}

	/**
	 * Process a call originating from the given request. Uses the
	 * {@link RPC#invokeAndEncodeResponse(Object, java.lang.reflect.Method, Object[])}
	 * method to do the actual work.
	 * 
	 * After processing it write response to outbound stream
	 * 
	 * @param message
	 * @param outbound
	 * @throws SerializationException
	 *             if we cannot serialize the response
	 * @throws SocketIOException
	 * @throws UnexpectedException
	 *             if the invocation throws a checked exception that is not
	 *             declared in the service method's signature
	 * @throws RuntimeException
	 *             if the service method throws an unchecked exception (the
	 *             exception will be the one thrown by the service)
	 */
	public void processRequest(String message, SocketIOOutbound outbound) throws SerializationException,
			SocketIOException {

		// Get request id
		String requestId = message.substring(0, 4);
		message = message.substring(4);

		// Result of encoding for writing
		String result = null;

		try {
			LOG.info("Message ID: "+requestId);
			RPCRequest rpcRequest = RPC.decodeRequest(message, delegate.getClass(), this);

			result = RPC.invokeAndEncodeResponse(delegate, rpcRequest.getMethod(),
					rpcRequest.getParameters(), rpcRequest.getSerializationPolicy(),
					rpcRequest.getFlags());
			LOG.info("Encoded result: Flags("+rpcRequest.getFlags()+") "+result);
		} catch (IncompatibleRemoteServiceException ex) {
			LOG.info("An IncompatibleRemoteServiceException was thrown while processing this call.", ex);
			result = RPC.encodeResponseForFailure(null, ex);
		} catch (RpcTokenException tokenException) {
			LOG.info("An RpcTokenException was thrown while processing this call.", tokenException);
			result = RPC.encodeResponseForFailure(null, tokenException);
		} finally {
			// Send the result message
			if (result != null)
				outbound.sendMessage("0" + requestId + result);
		}
	}

	/**
	 * Push the message to subscribers by given node
	 * 
	 * @param methodName
	 * @param node
	 * @param result
	 */
	protected <T> void publishMessage(Method method, T result, String node) {
		LOG.info("Publish message to node: "+node);
		dispatcher.publish(Topic.topic(node), new Object[] { method, result, threadSocket.get() });
	}

	/**
	 * Add the subscriber to given nodes
	 * 
	 * @param methodName
	 * @param node
	 */
	protected void addSubscriber(String... nodes) {
		if (nodes != null && nodes.length > 0) {
			LOG.info("Nodes subscribed: "+nodes);
			dispatcher.subscribe(Topics.anyOf(nodes), Object.class, threadSocket.get());
		}
	}

	public final SerializationPolicy getSerializationPolicy(String moduleBaseURL, String strongName) {
		// Try to get policy from cache
		SerializationPolicy serializationPolicy = getCachedSerializationPolicy(moduleBaseURL, strongName);

		// Policy found in cache, return it
		if (serializationPolicy != null) {
			return serializationPolicy;
		}

		// Load policy and store it in cache
		serializationPolicy = doGetSerializationPolicy(moduleBaseURL, strongName);

		if (serializationPolicy == null) {
			// Failed to get the requested serialization policy; use the default
			log("WARNING: Failed to get the SerializationPolicy '"
					+ strongName
					+ "' for module '"
					+ moduleBaseURL
					+ "'; a legacy, 1.3.3 compatible, serialization policy will be used.  You may experience SerializationExceptions as a result.");
			serializationPolicy = RPC.getDefaultSerializationPolicy();
		}

		// This could cache null or an actual instance. Either way we will not
		// attempt to lookup the policy again.
		putCachedSerializationPolicy(moduleBaseURL, strongName, serializationPolicy);

		return serializationPolicy;
	}

	/**
	 * Gets the {@link SerializationPolicy} for given module base URL and strong
	 * name if there is one.
	 * 
	 * Override this method to provide a {@link SerializationPolicy} using an
	 * alternative approach.
	 * 
	 * @param moduleBaseURL
	 *            as specified in the incoming payload
	 * @param strongName
	 *            a strong name that uniquely identifies a serialization policy
	 *            file
	 * @return a {@link SerializationPolicy} for the given module base URL and
	 *         strong name, or <code>null</code> if there is none
	 */
	protected SerializationPolicy doGetSerializationPolicy(String moduleBaseURL, String strongName) {
		return loadSerializationPolicy(this, moduleBaseURL, strongName);
	}

	private SerializationPolicy getCachedSerializationPolicy(String moduleBaseURL, String strongName) {
		synchronized (serializationPolicyCache) {
			return serializationPolicyCache.get(moduleBaseURL + strongName);
		}
	}

	private void putCachedSerializationPolicy(String moduleBaseURL, String strongName,
			SerializationPolicy serializationPolicy) {
		synchronized (serializationPolicyCache) {
			serializationPolicyCache.put(moduleBaseURL + strongName, serializationPolicy);
		}
	}
	
	/**
	 * Initialize subscribers when new connection
	 */
	protected abstract void initSubscribers();
}
